package DianShang_2024.ds_03.feature

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer, StringIndexerModel, VectorAssembler}
import org.apache.spark.ml.linalg.{SparseVector, Vectors}
import org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions.{array, col, collect_list, first, lit, max, substring, udf, when}
import spire.compat.fractional
import org.apache.spark.sql.{SparkSession, functions => F}
import org.apache.spark.ml.functions.vector_to_array

import java.util.Properties

object feature_count02 {
  def main(args: Array[String]): Unit = {
    /*
          据Hive的dwd库中相关表或MySQL数据库shtd_store中订单相关表（order_detail、order_info、sku_info），对用户购买过的商品进行去重，
          将其转换为以下格式：第一列为用户id mapping，第二列为用户购买过的商品id mapping，按照user_id与sku_id进行升序排序，输出前5行，将结
          果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("特征工程01")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      //      .config("spark.serializer","org.apache.spark.serializer")
      //      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._


    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    def read_mysql(mysql_name:String):Unit={
      val mysql_jdbc="jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false"
      spark.read.jdbc(mysql_jdbc,mysql_name,connect)
        .createOrReplaceTempView(mysql_name)
    }

    read_mysql("sku_info")
    read_mysql("order_info")
    read_mysql("order_detail")

    /*
          1.剔除订单信息表与订单详细信息表中用户id与商品id不存在于现有的维表中的记录(就是将三张表连接起来，如果有用户一次都没有买过，那么就连接不起来
         也就达到了剔除的效果)
         2.对用户购买过的商品进行去重
         3.where t1.user_id is not null and t3.sku_id is not null这一条要加，可以起到筛选的效果帮助剔除
     */
    val r1=spark.sql(
      """
        |select distinct
        |t1.user_id,
        |t3.id
        |from order_info as t1
        |join order_detail as t2
        |on t2.order_id=t1.id
        |join sku_info as t3
        |on t3.id=t2.sku_id
        |where t1.user_id is not null and t3.id is not null
        |""".stripMargin)

    //    r1.show

    //  定义StringIndexer用于映射(mapping),每一列都需要定义
    //  这里从机器学习的角度来说是注册模型，也就是测试模型
    val user_indexer=new StringIndexer()
      .setInputCol("user_id")                               //  选择改变的字段
      .setOutputCol("user_id_mapping")                      //  输出的字段
      .fit(r1)                                            //  根据r1来选择字段


    val sku_indexer=new StringIndexer()
      .setInputCol("id")
      .setOutputCol("sku_id_mapping")
      .fit(r1)


    //  执行映射(mapping)
    //  然后这里从机器学习的角度来说就是根据测试模型来训练模型
    val mapping_01=user_indexer.transform(r1)
    val result=sku_indexer.transform(mapping_01)


    println("-------user_id_mapping与sku_id_mapping数据前5条如下：-------")
    //  设置排序和选择字段
    result
      .withColumn("user_id_mapping",col("user_id_mapping").cast("int"))
      .withColumn("sku_id_mapping",col("sku_id_mapping").cast("int"))
      .select("user_id_mapping","sku_id_mapping")                         //  选择字段
      .orderBy("user_id_mapping","sku_id_mapping")                  //  排序规则
      .show(5,false)




    //      ______________第二题________________________
    /*
          根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，按照用户id进
          行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下

          one_hot编码:一行数据里面只有1和0,而且使用one-hot编码之前也要先映射，在one-hot
     */

    val result01=result.select("user_id_mapping","sku_id_mapping")

//  ---------------------------------------------------------------------------------
    // 创建VectorAssembler来组合特征列
    val assembler = new VectorAssembler()
      .setInputCols(Array("sku_id_mapping"))
      .setOutputCol("features")

    // 创建OneHotEncoder来进行one-hot编码
    val encoder = new OneHotEncoder()
      .setInputCol("features")
      .setOutputCol("sku_id_onehot")


    // 创建Pipeline来执行上述操作
    val pipeline = new Pipeline().setStages(Array(assembler, encoder))
    val model = pipeline.fit(result01)
    val oneHotUserSkuDF = model.transform(result01)

    oneHotUserSkuDF.show(false)

    // 对用户进行聚合，以获取one-hot编码的商品信息
    val aggregatedDF = oneHotUserSkuDF.groupBy("user_id_mapping").agg(F.max("sku_id_onehot").alias("sku_features"))
    // 对结果进行逆向操作，拆分one-hot编码的商品信息列
    val numDistinctSkuIds = oneHotUserSkuDF.select("sku_id_mapping").distinct().count.toInt

//    val resultDF = aggregatedDF.selectExpr("user_id_mapping",
//      (0 until numDistinctSkuIds).map(i => s"CAST(sku_features[$i] AS DOUBLE) AS sku_id$i"):_*
//    )
//
//    // 输出第一行前5列数据
//    resultDF.limit(5).show(truncate = false)
//  -----------------------------------------------------------------------------------------------------

//  对sku字段进行索引
//      val stringender:StringIndexerModel=new StringIndexer()
//        .setInputCol("sku_id_mapping")
//        .setOutputCol("sku_id_index")
//        .fit(result01)
//
//    val index_data=stringender.transform(result01)
//
//    //  使用oneHotEncoder对特征进行one-hot编码
//    val encoder:OneHotEncoder=new OneHotEncoder()
//      .setInputCol("sku_id_index")
//      .setOutputCol("sku_id_onehot")
//      .setDropLast(false)
//
//    val encoder_data=encoder.fit(index_data).transform(index_data)
//    encoder_data.show
//
//    //  进行向量拆分
//    val result02=encoder_data.withColumn("result",vector_to_array(col("sku_id_onehot"))).show(false)
//







    spark.close()

  }

}
